package com.suxg.hbase.test;

import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;

import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;

import com.suxg.hbase.api.RowMapper;
import com.suxg.hbase.mapper.StudentMapper;
import com.suxg.hbase.model.F;
import com.suxg.hbase.model.StudentModel;
import com.suxg.hbase.utils.HBaseUtilsFilters;
import com.suxg.hbase.utils.HBaseUtilsMapper;
import com.suxg.hbase.utils.SortListUtils;

public class HBaseUtilsTest4 {

	private static final String TABLE_NAME = "class";
	private static final String TEACHER = "teacher";
	private static final String STUDENT = "student";
	private static final String ROWKEY = "rowKey1";

	public static void main(String[] args) throws Exception {
//   	createTable();
//    	insertData();
//		getRow();
//		getScanner("oldp_ns:test2", "20_20210422 ", "F");
//		getClassScanner("class", "rowKey1", "student",new StudentMapper(),new StudentModel() );

//		getFScanner("oldp_ns:test2", "20_20210422", "F",new FMapper());
//    	getCell();
//		getScanner();
//		getScannerWithFilter();
		// 获取单个RowKey 所有数据
		StudentModel s = (StudentModel) HBaseUtilsFilters.getRow(TABLE_NAME, ROWKEY, STUDENT, new StudentModel());
		// 获取指定Rowkey, 列族指定列的值
		String signValue = HBaseUtilsFilters.getCell(TABLE_NAME, ROWKEY, STUDENT, "name");
		// 扫描全表
		List<StudentModel> list = HBaseUtilsFilters.ScannerList(TABLE_NAME, StudentModel.class);
		
		//JAVA 8 流排序 
		List<StudentModel> students = list.stream().

				sorted((Comparator.comparingInt(StudentModel::getAge).reversed()))

				.collect(Collectors.toList());
		//单个字段升序排序
		List<StudentModel> signList= (List<StudentModel>) SortListUtils.sort(list, "age");
		
		
		//多字段排序 fields 里的值为要排序的字段, sorts 里为排序的方法与fields里的对应，为null默认生序
		String [] fields= {"age","gender"};
		String[] sorts={SortListUtils.DESC,SortListUtils.ASC};
		List<StudentModel> 	sortList=(List<StudentModel>) SortListUtils.sort(list, fields, sorts);
		
		
		
		

		
		System.out.println(s.getAge() + "###获取单个RowKey###" + s.getName());

		System.out.println("#####获取单个值###" + signValue);

		System.out.println(list.size());
		for (StudentModel sm : list) {
			System.out.println(sm.getName() + "------" + sm.getAge() + "------" + sm.getGender());
		}

		for (StudentModel sm : students) {
			System.out.println("###JAVA 8 流排序 测试#"+sm.getName() + "------" + sm.getAge() + "------" + sm.getGender());
		}
		
		for (StudentModel sm : signList) {
			System.out.println("###单个字段升序排序 测试#"+sm.getName() + "------" + sm.getAge() + "------" + sm.getGender());
		}
		
		for (StudentModel sm : sortList) {
			System.out.println("###sortList排序测试#"+sm.getName() + "------" + sm.getAge() + "------" + sm.getGender());
		}
	}

	/**
	 * 创建表
	 */
	public static void createTable() {
		// 新建表
		List<String> columnFamilies = Arrays.asList(TEACHER, STUDENT);
		boolean table = HBaseUtilsMapper.createTable(TABLE_NAME, columnFamilies);
	}

	/**
	 * 插入表
	 */
	public static void insertData() {
		List<Pair<String, String>> pairs1 = Arrays.asList(new Pair<>("name", "Tom"), new Pair<>("age", "22"),
				new Pair<>("gender", "1"));
		HBaseUtilsMapper.putRow(TABLE_NAME, "rowKey1", STUDENT, pairs1);

		List<Pair<String, String>> pairs2 = Arrays.asList(new Pair<>("name", "Jack"), new Pair<>("age", "33"),
				new Pair<>("gender", "2"));
		HBaseUtilsMapper.putRow(TABLE_NAME, "rowKey2", STUDENT, pairs2);

		List<Pair<String, String>> pairs3 = Arrays.asList(new Pair<>("name", "Mike"), new Pair<>("age", "44"),
				new Pair<>("gender", "1"));
		HBaseUtilsMapper.putRow(TABLE_NAME, "rowKey3", STUDENT, pairs3);
	}

	/**
	 * 按rowkey 取出列族下某字段值
	 */
	public static void getRow() {
		StudentModel result = HBaseUtilsMapper.getRow(TABLE_NAME, "rowKey1", new StudentMapper());
		if (result != null) {
			System.out.println(result.getAge() + "#####name=" + result.getName());
		}
		System.out.println("query hbase shell:" + "get 'class','rowKey1',{COLUMN => 'student:name'} ");

	}

	// 同上
	public static void getCell() {
		String cell = HBaseUtilsMapper.getCell(TABLE_NAME, "rowKey2", STUDENT, "age");
		System.out.println("cell age :" + cell);

	}

	// 全表查询
	public static void getScanner() {
//		long starttime=System.currentTimeMillis();
		List<StudentModel> list = HBaseUtilsMapper.getScanner(TABLE_NAME, new StudentMapper());
		long endtime=System.currentTimeMillis();
//		System.out.println("$$$$$$$$$$$$$"+(endtime-starttime)/1000);
//		for (StudentModel sm : list) {
//			System.out.println(sm.getName() + sm.getAge() + sm.getGender());
//		}

	}

	public static void getFScanner(String tableName, String rowKey, String fimeys, RowMapper rowMappers, Object o) {
		List<F> list = HBaseUtilsMapper.getHbaseObjectList(tableName, rowMappers, rowKey, fimeys, o);
		for (F f : list) {
			System.out.println(f.getCid() + "###f###" + f.getSdf());
		}

	}

	public static void getClassScanner(String tableName, String rowKey, String fimeys, RowMapper rowMappers, Object o) {
		List<StudentModel> list = HBaseUtilsMapper.getHbaseObjectList(tableName, rowMappers, rowKey, fimeys, o);
		System.out.println(list.size());
		for (StudentModel s : list) {
			System.out.println(s.getAge() + "###u###" + s.getName());
		}

	}

	public static void getScannerWithFilter() {
		FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
		SingleColumnValueFilter nameFilter = new SingleColumnValueFilter(Bytes.toBytes(STUDENT), Bytes.toBytes("name"),
				CompareOp.EQUAL, Bytes.toBytes("Jack"));
		filterList.addFilter(nameFilter);
		ResultScanner scanner = HBaseUtilsMapper.getScanner(TABLE_NAME, filterList);
		if (scanner != null) {
			scanner.forEach(result -> System.out.println(Bytes.toString(result.getRow()) + "->"
					+ Bytes.toString(result.getValue(Bytes.toBytes(STUDENT), Bytes.toBytes("name")))));
			scanner.close();
		}
	}

	public static void deleteColumn() {
		boolean b = HBaseUtilsMapper.deleteColumn(TABLE_NAME, "rowKey2", STUDENT, "age");
		System.out.println("删除结果: " + b);
	}

	public static void deleteRow() {
		boolean b = HBaseUtilsMapper.deleteRow(TABLE_NAME, "rowKey2");
		System.out.println("删除结果: " + b);
	}

	public static void deleteTable() {
		boolean b = HBaseUtilsMapper.deleteTable(TABLE_NAME);
		System.out.println("删除结果: " + b);
	}
}
